package com.hellomaker.redismq.starter.config.listener;

import com.hellomaker.redismq.consume.GroupConsumer;
import com.hellomaker.redismq.consume.listener.ListenerContainer;
import com.hellomaker.redismq.core.channel.DataCodecStreamChannel;
import com.hellomaker.redismq.core.mq.SimpleRedisMQClient;
import com.hellomaker.redismq.starter.config.DoListen;
import com.hellomaker.redismq.starter.config.MessageListener;
import io.lettuce.core.StreamMessage;

import java.lang.reflect.Method;
import java.util.stream.Collectors;

public class ListenerMethod implements DoListen {

    SimpleRedisMQClient client;
    ListenerContainer<String, String> listenerContainer;
    MessageListener messageListener;
    Method proxyMethod;
    Object sourceObject;

    public ListenerMethod(SimpleRedisMQClient client, ListenerContainer<String, String> listenerContainer, MessageListener messageListener, Method proxyMethod, Object sourceObject) {
        this.client = client;
        this.listenerContainer = listenerContainer;
        this.messageListener = messageListener;
        this.proxyMethod = proxyMethod;
        this.sourceObject = sourceObject;
    }

    @Override
    public void doListen() {
//        listenerMethods.add(declaredMethod);
//        System.out.println("listener : " + messageListener);
        String groupName = messageListener.groupName();
        String channelName = messageListener.channelName();
        boolean autoAck = messageListener.isAutoAck();
        int maxLength = messageListener.maxLength();
        boolean exceptionJudge = messageListener.simpleExceptionJudge();
        DataCodecStreamChannel<String> channel = client.declareChannel(channelName);
        GroupConsumer<String, String> consumer = channel.declareGroupConsumer(groupName);
        Class<?>[] parameterTypes = proxyMethod.getParameterTypes();
        listenerContainer.receive(consumer, (consumer1, streamMessages) -> {
            Object[] args = new Object[parameterTypes.length];
            int index = 0;
            for (Class<?> parameterType : parameterTypes) {
                if (parameterType.equals(consumer1.getClass()) || parameterType.isAssignableFrom(consumer1.getClass())) {
                    args[index] = consumer1;
                } else if (parameterType.equals(streamMessages.getClass()) || parameterType.isAssignableFrom(streamMessages.getClass())) {
                    args[index] = streamMessages;
                }
                index++;
            }
            try {
                proxyMethod.invoke(sourceObject, args);
                if (!autoAck && exceptionJudge) {
                    consumer1.ack(streamMessages.stream().map(StreamMessage::getId).collect(Collectors.toList()));
                }
            } catch (Exception e) {
                if (!autoAck && exceptionJudge) {
                    consumer1.nack(streamMessages.stream().map(StreamMessage::getId).collect(Collectors.toList()));
                }
                e.printStackTrace();
            }
        }, autoAck, maxLength);
    }
}
